package com.sugon.ohdfs.integration.flink.job.stream.sink;

import com.sugon.ohdfs.integration.flink.domain.TestItem;
import com.sugon.ohdfs.integration.flink.util.DatetimeUtil;
import lombok.Data;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.parquet.avro.ParquetAvroWriters;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.springframework.beans.factory.annotation.Value;

import java.io.Serializable;

public class BulkParquetStreamSinkJob extends AbstractStreamSinkJob{

    @Value("${hdfs.base-path:hdfs://10.11.8.29:9001/flink/sinkTest/stream/sink/bulk/parquet}")
    String basePath;

    @Override
    protected Serializable transform(TestItem item) {
        return new TestItemParquet(item);
    }

    @Override
    protected StreamingFileSink buildSink() {
        return StreamingFileSink
                .<TestItemParquet>forBulkFormat(new Path(basePath), ParquetAvroWriters.forReflectRecord(TestItemParquet.class))
                .build();
    }

    /**
     * ParquetWriter无法识别localDatetime，所以需要转换一下，否则会导致文件一直处于inprogress无法complete
     */
    @Data
    public  static class TestItemParquet implements Serializable{

        public TestItemParquet(){}

        String name;
        String value;
        Long produceTimestamp;
        Long processTimeStamp;

        public TestItemParquet(TestItem item){
            this.name = item.getName();
            this.value = item.getValue();
            this.produceTimestamp = DatetimeUtil.getEpochMilli(item.getProduceTimestamp());
            this.processTimeStamp = DatetimeUtil.getEpochMilli(item.getProcessTimeStamp());
        }

        public TestItem toTestItem(){
            TestItem result = new TestItem();

            result.setName(name);
            result.setValue(value);
            if(produceTimestamp!=null && produceTimestamp>0){
                result.setProduceTimestamp(DatetimeUtil.getLocalDatetime(produceTimestamp));
            }
            if(processTimeStamp!=null && processTimeStamp>0){
                result.setProcessTimeStamp(DatetimeUtil.getLocalDatetime(processTimeStamp));
            }

            return result;
        }
    }
}
